package com.dreams.dream.common.async.disruptor;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmax.disruptor.ClaimStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.MultiThreadedClaimStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;

@SuppressWarnings("rawtypes")
public class DisruptorFactory implements EventFactory, Startable {

    Logger logger = LoggerFactory.getLogger(this.getClass());

    public final static String module = DisruptorFactory.class.getName();
    private Map<String, Disruptor> topicDisruptors;
    private String RingBufferSize;

    public DisruptorFactory(String ringBufferSize) {
        this.RingBufferSize = ringBufferSize;
    }

    @SuppressWarnings({ "unchecked" })
    public DisruptorFactory() {
        // TODO configure in xml
        this.RingBufferSize = "8";
        this.topicDisruptors = new ConcurrentHashMap();
    }

    @SuppressWarnings({"unchecked"})
    private Disruptor createDw() {
        // executorService = Executors.newFixedThreadPool(100);
        WaitStrategy waitStrategy = new YieldingWaitStrategy();
        ClaimStrategy claimStrategy = new MultiThreadedClaimStrategy(Integer.parseInt(RingBufferSize));
        return new Disruptor(this, Executors.newCachedThreadPool(), claimStrategy, waitStrategy);
    }

    /**
     * one event one EventDisruptor
     * 
     * @param topic
     * @return
     */
    @SuppressWarnings({ "unchecked" })
    public Disruptor createDisruptor(DomainEventHandler handler, String topic) {

        // 如果存在对应的disruptors,则取出直接返回
        if (topicDisruptors.containsKey(topic)) {
            return topicDisruptors.get(topic);
        }

        Disruptor disruptor = createDw();
        DomainEventHandlerAdapter dea = new DomainEventHandlerAdapter(handler);
        disruptor.handleEventsWith(dea);

        // 将disruptors对象放入缓存
        topicDisruptors.put(topic, disruptor);

        disruptor.start();

        return disruptor;
    }

    public EventDisruptor newInstance() {
        return new EventDisruptor();
    }

    @Override
    public void start() {
        // TODO Auto-generated method stub

    }

    @Override
    public void stop() {
        this.RingBufferSize = null;

    }
}
